package org.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.atguigu.gmall.realtime.annotation.NotSink;
import org.atguigu.gmall.realtime.bean.KeywordBean;
import org.atguigu.gmall.realtime.bean.TableProcess;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.sink.PhoenixSink;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Date;
import java.util.Properties;
import java.util.Random;
import java.util.stream.Collectors;


/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.util
 * Author: ZengHaiFeng
 * CreateTime:2023/4/22 11:18:21
 * Description: Todo
 * Version：1.0
 */
public class FlinkSinkUtil {
    /**
     * @description: build a phoenix sink.
     * @author: zenghaifeng
     * @date: 2023/4/23 16:42
     * @return SinkFunction<Tuple2<TableProcess>>
     **/
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink(){
        return new PhoenixSink();
    }

    /**
     * @description: build a kafka sink.
     * @author: zenghaifeng
     * @date: 2023/4/23 16:43
     * @param topic wanner output of the data into topic of the Kafka.
     * @return Sink<JSONObject>
     **/
    public static Sink<String> getKafkaSink(String topic) {
        return KafkaSink.<String>builder()
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setTransactionalIdPrefix(topic + new Random().nextInt(1000))
                .setBootstrapServers(Constant.KAFKA_SERVERS)
                .setRecordSerializer(
                        KafkaRecordSerializationSchema
                        .<String>builder()
                                .setTopic(topic)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build())
                .setProperty("transaction.timeout.ms",String.valueOf(15 * 60 * 1000))
                .build();
    }

    /**
     * @Author: zenghaifeng
     * @Date: 2023/4/24 20:24
     * @TODO: TODO building a Kafka sink function old means
     * @param: topic
     * @return: SinkFunction<java.lang.String>
     */
    public static SinkFunction<String> getKafkaSinkFunction(String topic){
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers",Constant.KAFKA_SERVERS);
        properties.setProperty("transaction.timeout.ms",String.valueOf(15 * 60 * 1000));
        return new FlinkKafkaProducer<String>("default", new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String s, @Nullable Long aLong) {
                return new ProducerRecord<>(topic,s.getBytes(StandardCharsets.UTF_8));
            }
        } ,properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static Sink<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        return KafkaSink.<Tuple2<JSONObject,TableProcess>>builder()
                .setBootstrapServers(Constant.KAFKA_SERVERS)
                .setRecordSerializer((KafkaRecordSerializationSchema<Tuple2<JSONObject, TableProcess>>)
                        (element, context, timestamp) -> new ProducerRecord<>(element.f1.getSinkTable(),
                        element.f0.toString().getBytes(StandardCharsets.UTF_8)))
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setTransactionalIdPrefix("Kafka_topic-")
                .setProperty("transaction.timeout.ms",String.valueOf(15 * 60 * 1000))
                .build();
    }

    public static <T> SinkFunction<T> getClickHouseSink(String table, Class<T> tClass) {
        /*
            wrap a clickhouse sink base on jdbc sin,
            because clickhouse is support  standard sql similar with jdbc
         */
        String columns = Arrays
                .stream(tClass.getDeclaredFields())
                .filter(field -> field.getAnnotation(NotSink.class) == null)
                .map(f -> CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, f.getName()))
                .collect(Collectors.joining(","));
        // build sql
        StringBuilder sql = new StringBuilder();
        sql
                .append("insert into ")
                .append(table)
                .append(" ( ")
                .append(columns)
                .append(" ) values (")
                .append(columns.replaceAll("[^,]+","?"))
                .append(" )");

        return JdbcSink.sink(
                sql.toString(),
                (JdbcStatementBuilder<T>) (ps, t) -> {
                    /*
                        suppose the sql was insert into user(a,b,c,d) values(?,?,?,?).
                        Then we will assignment to their four placeholders.
                        First step ,get all the fields through reflect.
                        Second step , get value of the fields that be used for assignment to their placeholders.
                     */
                    Field[] fields = tClass.getDeclaredFields();
                    try {
                        for (int i = 0,position = 1 ; i < fields.length; i++) {
                            Field field = fields[i];
                            if (field.getAnnotation(NotSink.class) == null){
                                // let private field's value to available, default is inaccessible..
                                field.setAccessible(true);
                                ps.setObject(position++, field.get(t));
                            }

                        }
                    }catch (IllegalAccessException e){
                        throw new RuntimeException(e);
                    }
                },
                new JdbcExecutionOptions
                        .Builder()
                        .withBatchSize(1000)
                        .withBatchIntervalMs(3000)
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withUsername("default")
                        .withPassword("aaaaaa")
                        .withUrl(Constant.CLICKHOUSE_URL)
                        .withDriverName(Constant.CLICKHOUSE_DRIVER)
                        .build());

    }
}




























